package com.atguigu.flink.datastreamapi.transform;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/4/4
 *
 * process算子是最底层的算子。
 *  map,filter,flatmap这些算子的功能都可以由process实现
 */
public class Demo1_Process
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("hadoop102", 8888)
                                                               .map(new WaterSensorMapFunction());


        //模拟process过滤
        ds.process(new ProcessFunction<WaterSensor, WaterSensor>()
        {
            /*
                WaterSensor value: 输入的数据
                Context ctx: 编程环境，可以从中获取丰富的工具
                Collector<WaterSensor> out： 输出收集器
             */
            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {
                //只留下s1传感器
                if ("s1".equals(value.getId())){
                    out.collect(value);
                }
            }
        })
          .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
